package com.tc.vms.iface;

import com.rabbitmq.client.Address;
import com.tc.vms.config.BrokerType;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.log.VmsClientLogger;
import com.tc.vms.message.MessageConst;
import com.tc.vms.producer.ChannelMetaData;
import com.tc.vms.utils.Utils;
import com.tc.vms.utils.VmsDomainAttr;
import com.tc.vms.utils.ZkUtils;
import org.slf4j.Logger;

import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * Created by yonghua.zhang on 2015/12/14.
 */
public abstract class VmsMetaData {
    protected final static Logger LOGGER = VmsClientLogger.getLog();
    protected final static String RETRY_QUEUE_PREFIX = "%RETRY%";
    protected final static String DEAD_LETTER_QUEUE_PREFIX = "%DLQ%";
    //broker connection string, brokerCluster maybe change when zk modified, but BrokerType mustn't be modified
    protected final String brokerCluster;
    protected final BrokerType brokerType;
    protected final VmsDomainAttr belongAttr;
    protected final Properties cfgCenterProps;
    protected final String origClientId;

    protected volatile int tpsLimit = 0;          // when > 0, will limit the speed,
    protected volatile boolean isSuspend = false; //if true, stop current thread
    protected volatile boolean isLoggerOpen = false;
    protected boolean isGrayMatched = true;
    protected final long createdTime = System.currentTimeMillis();

    public VmsMetaData(VmsDomainAttr belongAttr, Properties props, String origClientId) {
        this.cfgCenterProps = props;
        this.brokerCluster = ZkUtils.getVmsBrokerCluster(cfgCenterProps);
        this.brokerType = BrokerType.getBrokerTypeByName(ZkUtils.getVmsBrokerType(cfgCenterProps));
        this.belongAttr = belongAttr;
        this.isSuspend = ZkUtils.getVmsIsSuspend(props);
        this.tpsLimit = ZkUtils.getVmsTpsLimit(cfgCenterProps);
        this.origClientId = (origClientId == null || origClientId.isEmpty()) ? Utils.randomStr() : origClientId;
        this.isLoggerOpen = ZkUtils.getVmsLoggerSwitcher(props);

        try {
            if (ZkUtils.hasVmsClientFilter(cfgCenterProps)) {
                Pattern pat = this.getVmsClientFilter();
                isGrayMatched = pat.matcher(MessageConst.HOSTNAME).matches();
            }
        } catch (VmsClientException e) {
            LOGGER.warn("VmsMetaData getVmsClientFilter error.", e);
            isGrayMatched = false;
        }

    }

    public boolean isLoggerOpen() {
        return isLoggerOpen;
    }

    public void setIsLoggerOpen(boolean isLoggerOpen) {
        this.isLoggerOpen = isLoggerOpen;
    }

    public Properties getCommonProps() {
        return ZkUtils.getVmsCommonProps(cfgCenterProps);
    }

    public Properties getControlProps() {
        return ZkUtils.getVmsControlProps(cfgCenterProps);
    }

    public Pattern getVmsClientFilter() throws VmsClientException {

        String pString = ZkUtils.getVmsClientFilter(cfgCenterProps);
        Matcher mat = ZkUtils.vmsClientFilterPat.matcher(pString);
        String validPStr = "";
        while (mat.find()) {
            String findStr = mat.group();
            findStr = findStr.replaceAll("\\*", "\\\\d*");
            validPStr = !validPStr.isEmpty() ? validPStr + "|" + findStr : findStr;
        }
        if (validPStr != null && !validPStr.isEmpty()) {
            return Pattern.compile(validPStr);
        }

        throw new VmsClientException(String.format("VmsClientFilter is invalid, please check [%s].", pString));
    }

    public abstract String getZnodeName();

    public abstract String getClientInstName();

    public String getOrigClientId() {
        return origClientId;
    }

    public boolean isChannelMetaData() {
        return (this instanceof ChannelMetaData);
    }

    public boolean isSuspend() {
        return isSuspend;
    }

    public void setIsSuspend(boolean isSuspend) {
        this.isSuspend = isSuspend;
    }

    public int getTpsLimit() {
        return tpsLimit;
    }

    public void setTpsLimit(int tpsLimit) {
        this.tpsLimit = tpsLimit;
    }

    public String getBrokerCluster() {
        return brokerCluster;
    }

    public Address[] getBrokerAddress() {
        return Address.parseAddresses(brokerCluster);
    }

    public BrokerType getBrokerType() {
        return brokerType;
    }

    public VmsDomainAttr getBelongAttr() {
        return belongAttr;
    }

    public long getCreatedTime() {
        return createdTime;
    }

    public Properties getCfgCenterProps() {
        return cfgCenterProps;
    }

    public String getUserName() {
        return ZkUtils.getVmsUserName(cfgCenterProps);
    }

    public String getUserPasswd() {
        return ZkUtils.getVmsUserPasswd(cfgCenterProps);
    }

    public String getVhost() {
        return ZkUtils.getVmsVhost(cfgCenterProps);
    }

    public String getCfgZkCluster() {
        return ZkUtils.getCfgZkCluster(cfgCenterProps);
    }

    public String getCfgZkType() {
        return ZkUtils.getVmsCfgZkType(cfgCenterProps);
    }

    public boolean isGrayMatched() {
        return isGrayMatched;
    }

    @Override
    public String toString() {
        return "VmsMetaData{" +
                "brokerCluster='" + brokerCluster + '\'' +
                ", brokerType=" + brokerType +
                ", belongAttr=" + belongAttr +
                ", cfgCenterProps=" + cfgCenterProps +
                ", isSuspend=" + isSuspend +
                ", tpsLimit=" + tpsLimit +
                ", isGrayMatched=" + isGrayMatched +
                ", createdTime=" + createdTime +
                '}';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        VmsMetaData that = (VmsMetaData) o;

        if (!brokerCluster.equals(that.brokerCluster)) return false;
        if (brokerType != that.brokerType) return false;
        if (!belongAttr.equals(that.belongAttr)) return false;
        return cfgCenterProps.equals(that.cfgCenterProps);

    }

    @Override
    public int hashCode() {
        int result = brokerCluster.hashCode();
        result = 31 * result + brokerType.hashCode();
        result = 31 * result + belongAttr.hashCode();
        result = 31 * result + cfgCenterProps.hashCode();
        return result;
    }
}
